// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <ktest/ktest.h>

#include <cstdint>
#include <cstring>
#include <memory>
#include <vector>

#include <nebula/core/array.h>
#include <nebula/array/builder_nested.h>
#include <nebula/core/chunked_array.h>
#include <turbo/utility/status.h>
#include <nebula/testing/builder.h>
#include <nebula/testing/ktest_util.h>
#include <nebula/types/type.h>
#include <turbo/base/checked_cast.h>

namespace nebula {

    // ----------------------------------------------------------------------
    // Struct tests

    void ValidateBasicStructArray(const StructArray *result,
                                  const std::vector<uint8_t> &struct_is_valid,
                                  const std::vector<char> &list_values,
                                  const std::vector<uint8_t> &list_is_valid,
                                  const std::vector<int> &list_lengths,
                                  const std::vector<int> &list_offsets,
                                  const std::vector<int32_t> &int_values) {
        ASSERT_EQ(4, result->length());
        ASSERT_OK(result->validate_full());

        auto list_char_arr = std::dynamic_pointer_cast<ListArray>(result->field(0));
        auto char_arr = std::dynamic_pointer_cast<Int8Array>(list_char_arr->values());
        auto int32_arr = std::dynamic_pointer_cast<Int32Array>(result->field(1));

        ASSERT_EQ(nullptr, result->get_field_by_name("nonexistent"));
        ASSERT_TRUE(list_char_arr->equals(result->get_field_by_name("list")));
        ASSERT_TRUE(int32_arr->equals(result->get_field_by_name("int")));

        ASSERT_EQ(0, result->null_count());
        ASSERT_EQ(1, list_char_arr->null_count());
        ASSERT_EQ(0, int32_arr->null_count());

        // List<char>
        ASSERT_EQ(4, list_char_arr->length());
        ASSERT_EQ(10, list_char_arr->values()->length());
        for (size_t i = 0; i < list_offsets.size(); ++i) {
            ASSERT_EQ(list_offsets[i], list_char_arr->raw_value_offsets()[i]);
        }
        for (size_t i = 0; i < list_values.size(); ++i) {
            ASSERT_EQ(list_values[i], char_arr->value(i));
        }

        // Int32
        ASSERT_EQ(4, int32_arr->length());
        for (size_t i = 0; i < int_values.size(); ++i) {
            ASSERT_EQ(int_values[i], int32_arr->value(i));
        }
    }

    TEST(StructArray, FromFieldNames) {
        std::shared_ptr<Array> a, b, c, array, expected;
        a = assert_array_from_json(int32(), "[4, null]");
        b = assert_array_from_json(utf8(), R"([null, "foo"])");
        std::vector<std::string> field_names = {"a", "b"};

        auto res = StructArray::create({a, b}, field_names);
        ASSERT_OK(res);
        array = *res;
        expected = assert_array_from_json(STRUCT({field("a", int32()), field("b", utf8())}),
                                          R"([{"a": 4, "b": null}, {"a": null, "b": "foo"}])");
        AssertArraysEqual(*array, *expected);

        // With non-zero offsets
        res =
                StructArray::create({a, b}, field_names, /*null_bitmap =*/nullptr, /*null_count =*/0,
                        /*offset =*/1);
        ASSERT_OK(res);
        array = *res;
        expected = assert_array_from_json(STRUCT({field("a", int32()), field("b", utf8())}),
                                          R"([{"a": null, "b": "foo"}])");
        AssertArraysEqual(*array, *expected);

        res =
                StructArray::create({a, b}, field_names, /*null_bitmap =*/nullptr, /*null_count =*/0,
                        /*offset =*/2);
        ASSERT_OK(res);
        array = *res;
        expected = assert_array_from_json(STRUCT({field("a", int32()), field("b", utf8())}), R"([])");
        AssertArraysEqual(*array, *expected);

        // Offset greater than length
        res =
                StructArray::create({a, b}, field_names, /*null_bitmap =*/nullptr, /*null_count =*/0,
                        /*offset =*/3);
        ASSERT_RAISES(turbo::StatusCode::kOutOfRange, res);

        // With null bitmap
        std::shared_ptr<Buffer> null_bitmap;
        BitmapFromVector<bool>({false, true}, &null_bitmap);
        res = StructArray::create({a, b}, field_names, null_bitmap);
        ASSERT_OK(res);
        array = *res;
        expected = assert_array_from_json(STRUCT({field("a", int32()), field("b", utf8())}),
                                          R"([null, {"a": null, "b": "foo"}])");
        AssertArraysEqual(*array, *expected);

        // Mismatching array lengths
        field_names = {"a", "c"};
        c = assert_array_from_json(int64(), "[1, 2, 3]");
        res = StructArray::create({a, c}, field_names);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, res);

        // Mismatching number of fields
        field_names = {"a", "b", "c"};
        res = StructArray::create({a, b}, field_names);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, res);

        // Fail on 0 children (cannot infer array length)
        field_names = {};
        res = StructArray::create({}, field_names);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, res);
    }

    TEST(StructArray, FromFields) {
        std::shared_ptr<Array> a, b, c, array, expected;
        std::shared_ptr<Field> fa, fb, fc;
        a = assert_array_from_json(int32(), "[4, 5]");
        b = assert_array_from_json(utf8(), R"([null, "foo"])");
        fa = field("a", int32(), /*nullable =*/false);
        fb = field("b", utf8(), /*nullable =*/true);
        fc = field("b", int64(), /*nullable =*/true);

        auto res = StructArray::create({a, b}, {fa, fb});
        ASSERT_OK(res);
        array = *res;
        expected =
                assert_array_from_json(STRUCT({fa, fb}), R"([{"a": 4, "b": null}, {"a": 5, "b": "foo"}])");
        AssertArraysEqual(*array, *expected);

        // Mismatching array lengths
        c = assert_array_from_json(int64(), "[1, 2, 3]");
        res = StructArray::create({a, c}, {fa, fc});
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, res);

        // Mismatching number of fields
        res = StructArray::create({a, b}, {fa, fb, fc});
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, res);

        // Fail on 0 children (cannot infer array length)
        res = StructArray::create({}, std::vector<std::shared_ptr<Field>>{});
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, res);
    }

    TEST(StructArray, Validate) {
        auto a = assert_array_from_json(int32(), "[4, 5]");
        auto type = STRUCT({field("a", int32())});
        auto children = std::vector<std::shared_ptr<Array>>{a};

        auto arr = std::make_shared<StructArray>(type, 2, children);
        ASSERT_OK(arr->validate_full());
        arr = std::make_shared<StructArray>(type, 1, children, nullptr, 0, /*offset=*/1);
        ASSERT_OK(arr->validate_full());
        arr = std::make_shared<StructArray>(type, 0, children, nullptr, 0, /*offset=*/2);
        ASSERT_OK(arr->validate_full());

        // Length + offset < child length, but it's ok
        arr = std::make_shared<StructArray>(type, 1, children, nullptr, 0, /*offset=*/0);
        ASSERT_OK(arr->validate_full());

        // Length + offset > child length
        arr = std::make_shared<StructArray>(type, 1, children, nullptr, 0, /*offset=*/2);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->validate_full());

        // Offset > child length
        arr = std::make_shared<StructArray>(type, 0, children, nullptr, 0, /*offset=*/3);
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->validate_full());
    }

    TEST(StructArray, Flatten) {
        auto type =
                STRUCT({field("a", int32()), field("b", utf8()), field("c", list(boolean()))});
        {
            auto struct_arr = std::static_pointer_cast<StructArray>(assert_array_from_json(
                    type, R"([[1, "a", [null, false]], [null, "bc", []], [2, null, null]])"));
            ASSERT_OK_AND_ASSIGN(auto flattened, struct_arr->Flatten(default_memory_pool()));
            AssertArraysEqual(*assert_array_from_json(int32(), "[1, null, 2]"), *flattened[0],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(utf8(), R"(["a", "bc", null])"), *flattened[1],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(list(boolean()), "[[null, false], [], null]"),
                              *flattened[2], /*verbose=*/true);
        }
        {
            ArrayVector children = {
                    assert_array_from_json(int32(), "[1, 2, 3, 4]")->slice(1, 3),
                    assert_array_from_json(utf8(), R"([null, "ab", "cde", null])")->slice(1, 3),
                    assert_array_from_json(list(boolean()), "[[true], [], [true, false, null], [false]]")
                            ->slice(1, 3),
            };

            // Without slice or top-level nulls
            auto struct_arr = std::make_shared<StructArray>(type, 3, children);
            ASSERT_OK_AND_ASSIGN(auto flattened, struct_arr->Flatten(default_memory_pool()));
            AssertArraysEqual(*assert_array_from_json(int32(), "[2, 3, 4]"), *flattened[0],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(utf8(), R"(["ab", "cde", null])"), *flattened[1],
                    /*verbose=*/true);
            AssertArraysEqual(
                    *assert_array_from_json(list(boolean()), "[[], [true, false, null], [false]]"),
                    *flattened[2], /*verbose=*/true);

            // With slice
            struct_arr = std::make_shared<StructArray>(type, 2, children, /*null_bitmap=*/nullptr,
                    /*null_count=*/0, /*offset=*/1);
            ASSERT_OK_AND_ASSIGN(flattened, struct_arr->Flatten(default_memory_pool()));
            AssertArraysEqual(*assert_array_from_json(int32(), "[3, 4]"), *flattened[0],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(utf8(), R"(["cde", null])"), *flattened[1],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(list(boolean()), "[[true, false, null], [false]]"),
                              *flattened[2], /*verbose=*/true);

            struct_arr = std::make_shared<StructArray>(type, 1, children, /*null_bitmap=*/nullptr,
                    /*null_count=*/0, /*offset=*/2);
            ASSERT_OK_AND_ASSIGN(flattened, struct_arr->Flatten(default_memory_pool()));
            AssertArraysEqual(*assert_array_from_json(int32(), "[4]"), *flattened[0],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(utf8(), R"([null])"), *flattened[1],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(list(boolean()), "[[false]]"), *flattened[2],
                    /*verbose=*/true);

            // With top-level nulls
            std::shared_ptr<Buffer> null_bitmap;
            BitmapFromVector<bool>({true, false, true}, &null_bitmap);
            struct_arr = std::make_shared<StructArray>(type, 3, children, null_bitmap);
            ASSERT_OK_AND_ASSIGN(flattened, struct_arr->Flatten(default_memory_pool()));
            AssertArraysEqual(*assert_array_from_json(int32(), "[2, null, 4]"), *flattened[0],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(utf8(), R"(["ab", null, null])"), *flattened[1],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(list(boolean()), "[[], null, [false]]"),
                              *flattened[2], /*verbose=*/true);

            // With slice and top-level nulls
            struct_arr = std::make_shared<StructArray>(type, 2, children, null_bitmap,
                    /*null_count=*/1, /*offset=*/1);
            ASSERT_OK_AND_ASSIGN(flattened, struct_arr->Flatten(default_memory_pool()));
            AssertArraysEqual(*assert_array_from_json(int32(), "[null, 4]"), *flattened[0],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(utf8(), R"([null, null])"), *flattened[1],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(list(boolean()), "[null, [false]]"), *flattened[2],
                    /*verbose=*/true);

            struct_arr = std::make_shared<StructArray>(type, 1, children, null_bitmap,
                    /*null_count=*/0, /*offset=*/2);
            ASSERT_OK_AND_ASSIGN(flattened, struct_arr->Flatten(default_memory_pool()));
            AssertArraysEqual(*assert_array_from_json(int32(), "[4]"), *flattened[0],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(utf8(), R"([null])"), *flattened[1],
                    /*verbose=*/true);
            AssertArraysEqual(*assert_array_from_json(list(boolean()), "[[false]]"), *flattened[2],
                    /*verbose=*/true);
        }
    }

    /// ARROW-7740: Flattening a slice shouldn't affect the parent array.
    TEST(StructArray, FlattenOfSlice) {
        auto a = assert_array_from_json(int32(), "[4, 5]");
        auto type = STRUCT({field("a", int32())});
        auto children = std::vector<std::shared_ptr<Array>>{a};

        auto arr = std::make_shared<StructArray>(type, 2, children);
        ASSERT_OK(arr->validate_full());

        auto slice = turbo::checked_pointer_cast<StructArray>(arr->slice(0, 1));
        ASSERT_OK(slice->validate_full());

        ASSERT_OK_AND_ASSIGN(auto flattened, slice->Flatten(default_memory_pool()));

        ASSERT_OK(slice->validate_full());
        ASSERT_OK(arr->validate_full());
    }

    TEST(StructArray, can_reference_field_by_name) {
        auto a = assert_array_from_json(int8(), "[4, 5]");
        auto b = assert_array_from_json(int16(), "[6, 7]");
        auto c = assert_array_from_json(int32(), "[8, 9]");
        auto d = assert_array_from_json(int64(), "[10, 11]");
        auto children = std::vector<std::shared_ptr<Array>>{a, b, c, d};

        auto f0 = field("f0", int8());
        auto f1 = field("f1", int16());
        auto f2 = field("f2", int32());
        auto f3 = field("f1", int64());
        auto type = STRUCT({f0, f1, f2, f3});

        auto arr = std::make_shared<StructArray>(type, 2, children);

        ASSERT_OK(arr->can_reference_field_by_name("f0"));
        ASSERT_OK(arr->can_reference_field_by_name("f2"));
        // Not found
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->can_reference_field_by_name("nope"));

        // Duplicates
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->can_reference_field_by_name("f1"));
    }

    TEST(StructArray, can_reference_fields_by_names) {
        auto a = assert_array_from_json(int8(), "[4, 5]");
        auto b = assert_array_from_json(int16(), "[6, 7]");
        auto c = assert_array_from_json(int32(), "[8, 9]");
        auto d = assert_array_from_json(int64(), "[10, 11]");
        auto children = std::vector<std::shared_ptr<Array>>{a, b, c, d};

        auto f0 = field("f0", int8());
        auto f1 = field("f1", int16());
        auto f2 = field("f2", int32());
        auto f3 = field("f1", int64());
        auto type = STRUCT({f0, f1, f2, f3});

        auto arr = std::make_shared<StructArray>(type, 2, children);

        ASSERT_OK(arr->can_reference_fields_by_names({"f0", "f2"}));
        ASSERT_OK(arr->can_reference_fields_by_names({"f2", "f0"}));

        // Not found
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->can_reference_fields_by_names({"nope"}));
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->can_reference_fields_by_names({"f0", "nope"}));
        // Duplicates
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->can_reference_fields_by_names({"f1"}));
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->can_reference_fields_by_names({"f0", "f1"}));
        // Both
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, arr->can_reference_fields_by_names({"f0", "f1", "nope"}));
    }

    // ----------------------------------------------------------------------------------
    // Struct test
    class TestStructBuilder : public ::testing::Test {
    public:
        void SetUp() {
            auto int32_type = int32();
            auto char_type = int8();
            auto list_type = list(char_type);

            std::vector<std::shared_ptr<DataType>> types = {list_type, int32_type};
            std::vector<std::shared_ptr<Field>> fields;
            fields.push_back(field("list", list_type));
            fields.push_back(field("int", int32_type));

            type_ = STRUCT(fields);
            value_fields_ = fields;

            std::unique_ptr<ArrayBuilder> tmp;
            ASSERT_OK(MakeBuilder(pool_, type_, &tmp));
            builder_.reset(turbo::checked_cast<StructBuilder *>(tmp.release()));
            ASSERT_EQ(2, static_cast<int>(builder_->num_fields()));
        }

        void Done() {
            std::shared_ptr<Array> out;
            FinishAndCheckPadding(builder_.get(), &out);
            result_ = std::dynamic_pointer_cast<StructArray>(out);
        }

    protected:
        std::vector<std::shared_ptr<Field>> value_fields_;

        MemoryPool *pool_ = default_memory_pool();
        std::shared_ptr<DataType> type_;
        std::shared_ptr<StructBuilder> builder_;
        std::shared_ptr<StructArray> result_;
    };

    TEST_F(TestStructBuilder, TestAppendNull) {
        ASSERT_OK(builder_->append_null());
        ASSERT_OK(builder_->append_null());
        ASSERT_EQ(2, static_cast<int>(builder_->num_fields()));

        Done();

        ASSERT_OK(result_->validate_full());

        ASSERT_EQ(2, static_cast<int>(result_->num_fields()));
        ASSERT_EQ(2, result_->length());
        ASSERT_EQ(2, result_->field(0)->length());
        ASSERT_EQ(2, result_->field(1)->length());
        ASSERT_TRUE(result_->is_null(0));
        ASSERT_TRUE(result_->is_null(1));
        ASSERT_EQ(0, result_->field(0)->null_count());
        ASSERT_EQ(0, result_->field(1)->null_count());

        ASSERT_EQ(Type::LIST, result_->field(0)->type_id());
        ASSERT_EQ(Type::INT32, result_->field(1)->type_id());
    }

    TEST_F(TestStructBuilder, TestBasics) {
        std::vector<int32_t> int_values = {1, 2, 3, 4};
        std::vector<char> list_values = {'j', 'o', 'e', 'b', 'o', 'b', 'm', 'a', 'r', 'k'};
        std::vector<int> list_lengths = {3, 0, 3, 4};
        std::vector<int> list_offsets = {0, 3, 3, 6, 10};
        std::vector<uint8_t> list_is_valid = {1, 0, 1, 1};
        std::vector<uint8_t> struct_is_valid = {1, 1, 1, 1};

        ListBuilder *list_vb = turbo::checked_cast<ListBuilder *>(builder_->field_builder(0));
        Int8Builder *char_vb = turbo::checked_cast<Int8Builder *>(list_vb->value_builder());
        Int32Builder *int_vb = turbo::checked_cast<Int32Builder *>(builder_->field_builder(1));
        ASSERT_EQ(2, static_cast<int>(builder_->num_fields()));

        NEBULA_EXPECT_OK(builder_->resize(list_lengths.size()));
        NEBULA_EXPECT_OK(char_vb->resize(list_values.size()));
        NEBULA_EXPECT_OK(int_vb->resize(int_values.size()));

        int pos = 0;
        for (size_t i = 0; i < list_lengths.size(); ++i) {
            ASSERT_OK(list_vb->append(list_is_valid[i] > 0));
            int_vb->unsafe_append(int_values[i]);
            for (int j = 0; j < list_lengths[i]; ++j) {
                char_vb->unsafe_append(list_values[pos++]);
            }
        }

        for (size_t i = 0; i < struct_is_valid.size(); ++i) {
            ASSERT_OK(builder_->append(struct_is_valid[i] > 0));
        }

        Done();

        ValidateBasicStructArray(result_.get(), struct_is_valid, list_values, list_is_valid,
                                 list_lengths, list_offsets, int_values);
    }

    TEST_F(TestStructBuilder, BulkAppend) {
        std::vector<int32_t> int_values = {1, 2, 3, 4};
        std::vector<char> list_values = {'j', 'o', 'e', 'b', 'o', 'b', 'm', 'a', 'r', 'k'};
        std::vector<int> list_lengths = {3, 0, 3, 4};
        std::vector<int> list_offsets = {0, 3, 3, 6};
        std::vector<uint8_t> list_is_valid = {1, 0, 1, 1};
        std::vector<uint8_t> struct_is_valid = {1, 1, 1, 1};

        ListBuilder *list_vb = turbo::checked_cast<ListBuilder *>(builder_->field_builder(0));
        Int8Builder *char_vb = turbo::checked_cast<Int8Builder *>(list_vb->value_builder());
        Int32Builder *int_vb = turbo::checked_cast<Int32Builder *>(builder_->field_builder(1));

        ASSERT_OK(builder_->resize(list_lengths.size()));
        ASSERT_OK(char_vb->resize(list_values.size()));
        ASSERT_OK(int_vb->resize(int_values.size()));

        ASSERT_OK(builder_->append_values(struct_is_valid.size(), struct_is_valid.data()));

        ASSERT_OK(list_vb->append_values(list_offsets.data(), list_offsets.size(),
                                         list_is_valid.data()));
        for (int8_t value: list_values) {
            char_vb->unsafe_append(value);
        }
        for (int32_t value: int_values) {
            int_vb->unsafe_append(value);
        }

        Done();
        ValidateBasicStructArray(result_.get(), struct_is_valid, list_values, list_is_valid,
                                 list_lengths, list_offsets, int_values);
    }

    TEST_F(TestStructBuilder, BulkAppendInvalid) {
        std::vector<int32_t> int_values = {1, 2, 3, 4};
        std::vector<char> list_values = {'j', 'o', 'e', 'b', 'o', 'b', 'm', 'a', 'r', 'k'};
        std::vector<int> list_lengths = {3, 0, 3, 4};
        std::vector<int> list_offsets = {0, 3, 3, 6};
        std::vector<uint8_t> list_is_valid = {1, 0, 1, 1};
        std::vector<uint8_t> struct_is_valid = {1, 0, 1, 1};  // should be 1, 1, 1, 1

        ListBuilder *list_vb = turbo::checked_cast<ListBuilder *>(builder_->field_builder(0));
        Int8Builder *char_vb = turbo::checked_cast<Int8Builder *>(list_vb->value_builder());
        Int32Builder *int_vb = turbo::checked_cast<Int32Builder *>(builder_->field_builder(1));

        ASSERT_OK(builder_->Reserve(list_lengths.size()));
        ASSERT_OK(char_vb->Reserve(list_values.size()));
        ASSERT_OK(int_vb->Reserve(int_values.size()));

        ASSERT_OK(builder_->append_values(struct_is_valid.size(), struct_is_valid.data()));

        ASSERT_OK(list_vb->append_values(list_offsets.data(), list_offsets.size(),
                                         list_is_valid.data()));
        for (int8_t value: list_values) {
            char_vb->unsafe_append(value);
        }
        for (int32_t value: int_values) {
            int_vb->unsafe_append(value);
        }

        Done();
        // Even null bitmap of the parent Struct is not valid, validate will ignore it.
        ASSERT_OK(result_->validate_full());
    }

    TEST_F(TestStructBuilder, TestEquality) {
        std::shared_ptr<Array> array, equal_array;
        std::shared_ptr<Array> unequal_bitmap_array, unequal_offsets_array,
                unequal_values_array;

        std::vector<int32_t> int_values = {101, 102, 103, 104};
        std::vector<char> list_values = {'j', 'o', 'e', 'b', 'o', 'b', 'm', 'a', 'r', 'k'};
        std::vector<int> list_lengths = {3, 0, 3, 4};
        std::vector<int> list_offsets = {0, 3, 3, 6};
        std::vector<uint8_t> list_is_valid = {1, 0, 1, 1};
        std::vector<uint8_t> struct_is_valid = {1, 1, 1, 1};

        std::vector<int32_t> unequal_int_values = {104, 102, 103, 101};
        std::vector<char> unequal_list_values = {'j', 'o', 'e', 'b', 'o',
                                                 'b', 'l', 'u', 'c', 'y'};
        std::vector<int> unequal_list_offsets = {0, 3, 4, 6};
        std::vector<uint8_t> unequal_list_is_valid = {1, 1, 1, 1};
        std::vector<uint8_t> unequal_struct_is_valid = {1, 0, 0, 1};

        ListBuilder *list_vb = turbo::checked_cast<ListBuilder *>(builder_->field_builder(0));
        Int8Builder *char_vb = turbo::checked_cast<Int8Builder *>(list_vb->value_builder());
        Int32Builder *int_vb = turbo::checked_cast<Int32Builder *>(builder_->field_builder(1));
        ASSERT_OK(builder_->Reserve(list_lengths.size()));
        ASSERT_OK(char_vb->Reserve(list_values.size()));
        ASSERT_OK(int_vb->Reserve(int_values.size()));

        // set up two equal arrays, one of which takes an unequal bitmap
        ASSERT_OK(builder_->append_values(struct_is_valid.size(), struct_is_valid.data()));
        ASSERT_OK(list_vb->append_values(list_offsets.data(), list_offsets.size(),
                                         list_is_valid.data()));
        for (int8_t value: list_values) {
            char_vb->unsafe_append(value);
        }
        for (int32_t value: int_values) {
            int_vb->unsafe_append(value);
        }

        FinishAndCheckPadding(builder_.get(), &array);

        ASSERT_OK(builder_->resize(list_lengths.size()));
        ASSERT_OK(char_vb->resize(list_values.size()));
        ASSERT_OK(int_vb->resize(int_values.size()));

        ASSERT_OK(builder_->append_values(struct_is_valid.size(), struct_is_valid.data()));
        ASSERT_OK(list_vb->append_values(list_offsets.data(), list_offsets.size(),
                                         list_is_valid.data()));
        for (int8_t value: list_values) {
            char_vb->unsafe_append(value);
        }
        for (int32_t value: int_values) {
            int_vb->unsafe_append(value);
        }

        ASSERT_OK(builder_->finish(&equal_array));

        ASSERT_OK(builder_->resize(list_lengths.size()));
        ASSERT_OK(char_vb->resize(list_values.size()));
        ASSERT_OK(int_vb->resize(int_values.size()));

        // set up an unequal one with the unequal bitmap
        ASSERT_OK(builder_->append_values(unequal_struct_is_valid.size(),
                                          unequal_struct_is_valid.data()));
        ASSERT_OK(list_vb->append_values(list_offsets.data(), list_offsets.size(),
                                         list_is_valid.data()));
        for (int8_t value: list_values) {
            char_vb->unsafe_append(value);
        }
        for (int32_t value: int_values) {
            int_vb->unsafe_append(value);
        }

        ASSERT_OK(builder_->finish(&unequal_bitmap_array));

        ASSERT_OK(builder_->resize(list_lengths.size()));
        ASSERT_OK(char_vb->resize(list_values.size()));
        ASSERT_OK(int_vb->resize(int_values.size()));

        // set up an unequal one with unequal offsets
        ASSERT_OK(builder_->append_values(struct_is_valid.size(), struct_is_valid.data()));
        ASSERT_OK(list_vb->append_values(unequal_list_offsets.data(),
                                         unequal_list_offsets.size(),
                                         unequal_list_is_valid.data()));
        for (int8_t value: list_values) {
            char_vb->unsafe_append(value);
        }
        for (int32_t value: int_values) {
            int_vb->unsafe_append(value);
        }

        ASSERT_OK(builder_->finish(&unequal_offsets_array));

        ASSERT_OK(builder_->resize(list_lengths.size()));
        ASSERT_OK(char_vb->resize(list_values.size()));
        ASSERT_OK(int_vb->resize(int_values.size()));

        // set up an unequal one with unequal values
        ASSERT_OK(builder_->append_values(struct_is_valid.size(), struct_is_valid.data()));
        ASSERT_OK(list_vb->append_values(list_offsets.data(), list_offsets.size(),
                                         list_is_valid.data()));
        for (int8_t value: unequal_list_values) {
            char_vb->unsafe_append(value);
        }
        for (int32_t value: unequal_int_values) {
            int_vb->unsafe_append(value);
        }

        ASSERT_OK(builder_->finish(&unequal_values_array));

        // Test array equality
        EXPECT_TRUE(array->equals(array));
        EXPECT_TRUE(array->equals(equal_array));
        EXPECT_TRUE(equal_array->equals(array));
        EXPECT_FALSE(equal_array->equals(unequal_bitmap_array));
        EXPECT_FALSE(unequal_bitmap_array->equals(equal_array));
        EXPECT_FALSE(unequal_bitmap_array->equals(unequal_values_array));
        EXPECT_FALSE(unequal_values_array->equals(unequal_bitmap_array));
        EXPECT_FALSE(unequal_bitmap_array->equals(unequal_offsets_array));
        EXPECT_FALSE(unequal_offsets_array->equals(unequal_bitmap_array));

        // Test range equality
        EXPECT_TRUE(array->range_equals(0, 4, 0, equal_array));
        EXPECT_TRUE(array->range_equals(3, 4, 3, unequal_bitmap_array));
        EXPECT_TRUE(array->range_equals(0, 1, 0, unequal_offsets_array));
        EXPECT_FALSE(array->range_equals(0, 2, 0, unequal_offsets_array));
        EXPECT_FALSE(array->range_equals(1, 2, 1, unequal_offsets_array));
        EXPECT_FALSE(array->range_equals(0, 1, 0, unequal_values_array));
        EXPECT_TRUE(array->range_equals(1, 3, 1, unequal_values_array));
        EXPECT_FALSE(array->range_equals(3, 4, 3, unequal_values_array));
    }

    TEST_F(TestStructBuilder, TestZeroLength) {
        // All buffers are null
        Done();
        ASSERT_OK(result_->validate_full());
    }

    TEST_F(TestStructBuilder, TestSlice) {
        std::shared_ptr<Array> array, equal_array;
        std::shared_ptr<Array> unequal_bitmap_array, unequal_offsets_array,
                unequal_values_array;

        std::vector<int32_t> int_values = {101, 102, 103, 104};
        std::vector<char> list_values = {'j', 'o', 'e', 'b', 'o', 'b', 'm', 'a', 'r', 'k'};
        std::vector<int> list_lengths = {3, 0, 3, 4};
        std::vector<int> list_offsets = {0, 3, 3, 6};
        std::vector<uint8_t> list_is_valid = {1, 0, 1, 1};
        std::vector<uint8_t> struct_is_valid = {1, 1, 1, 1};

        ListBuilder *list_vb = turbo::checked_cast<ListBuilder *>(builder_->field_builder(0));
        Int8Builder *char_vb = turbo::checked_cast<Int8Builder *>(list_vb->value_builder());
        Int32Builder *int_vb = turbo::checked_cast<Int32Builder *>(builder_->field_builder(1));
        ASSERT_OK(builder_->Reserve(list_lengths.size()));
        ASSERT_OK(char_vb->Reserve(list_values.size()));
        ASSERT_OK(int_vb->Reserve(int_values.size()));

        ASSERT_OK(builder_->append_values(struct_is_valid.size(), struct_is_valid.data()));
        ASSERT_OK(list_vb->append_values(list_offsets.data(), list_offsets.size(),
                                         list_is_valid.data()));
        for (int8_t value: list_values) {
            char_vb->unsafe_append(value);
        }
        for (int32_t value: int_values) {
            int_vb->unsafe_append(value);
        }
        FinishAndCheckPadding(builder_.get(), &array);

        std::shared_ptr<StructArray> slice, slice2;
        std::shared_ptr<Int32Array> int_field;
        std::shared_ptr<ListArray> list_field;

        slice = std::dynamic_pointer_cast<StructArray>(array->slice(2));
        slice2 = std::dynamic_pointer_cast<StructArray>(array->slice(2));
        ASSERT_EQ(array->length() - 2, slice->length());

        ASSERT_TRUE(slice->equals(slice2));
        ASSERT_TRUE(array->range_equals(2, slice->length(), 0, slice));

        int_field = std::dynamic_pointer_cast<Int32Array>(slice->field(1));
        ASSERT_EQ(int_field->length(), slice->length());
        ASSERT_EQ(int_field->value(0), 103);
        ASSERT_EQ(int_field->value(1), 104);
        ASSERT_EQ(int_field->null_count(), 0);
        list_field = std::dynamic_pointer_cast<ListArray>(slice->field(0));
        ASSERT_FALSE(list_field->is_null(0));
        ASSERT_FALSE(list_field->is_null(1));
        ASSERT_EQ(list_field->value_length(0), 3);
        ASSERT_EQ(list_field->value_length(1), 4);
        ASSERT_EQ(list_field->null_count(), 0);

        slice = std::dynamic_pointer_cast<StructArray>(array->slice(1, 2));
        slice2 = std::dynamic_pointer_cast<StructArray>(array->slice(1, 2));
        ASSERT_EQ(2, slice->length());

        ASSERT_TRUE(slice->equals(slice2));
        ASSERT_TRUE(array->range_equals(1, 3, 0, slice));

        int_field = std::dynamic_pointer_cast<Int32Array>(slice->field(1));
        ASSERT_EQ(int_field->length(), slice->length());
        ASSERT_EQ(int_field->value(0), 102);
        ASSERT_EQ(int_field->value(1), 103);
        ASSERT_EQ(int_field->null_count(), 0);
        list_field = std::dynamic_pointer_cast<ListArray>(slice->field(0));
        ASSERT_TRUE(list_field->is_null(0));
        ASSERT_FALSE(list_field->is_null(1));
        ASSERT_EQ(list_field->value_length(0), 0);
        ASSERT_EQ(list_field->value_length(1), 3);
        ASSERT_EQ(list_field->null_count(), 1);
    }

    TEST(TestFieldRef, GetChildren) {
        auto struct_array = assert_array_from_json(STRUCT({field("a", float64())}), R"([
        {"a": 6.125},
        {"a": 0.0},
        {"a": -1}
      ])");

        ASSERT_OK_AND_ASSIGN(auto a, FieldRef("a").get_one(*struct_array));
        auto expected_a = assert_array_from_json(float64(), "[6.125, 0.0, -1]");
        AssertArraysEqual(*a, *expected_a);

        // more nested:
        struct_array =
                assert_array_from_json(STRUCT({field("a", STRUCT({field("a", float64())}))}), R"([
        {"a": {"a": 6.125}},
        {"a": {"a": 0.0}},
        {"a": {"a": -1}}
      ])");

        ASSERT_OK_AND_ASSIGN(a, FieldRef("a", "a").get_one(*struct_array));
        expected_a = assert_array_from_json(float64(), "[6.125, 0.0, -1]");
        AssertArraysEqual(*a, *expected_a);
    }

}  // namespace nebula
